package cn.virens.cmpt.rabbit;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import cn.hutool.core.util.IdUtil;
import cn.virens.cmpt.rabbit.packet.Packet;
import cn.virens.cmpt.rabbit.packet.PacketReply;
import cn.virens.main.AppConfig;
import cn.virens.main.cmpt.Component;

public class RabbitManagerImpl implements RabbitManager, Component {
	protected static final Logger logger = LoggerFactory.getLogger(RabbitManagerImpl.class);
	protected static final Charset charset = Charset.forName("UTF-8");

	protected final List<RabbitConsumerHandler<?>> mHandlers = new ArrayList<>();
	protected final String queue = "ludeng." + IdUtil.fastSimpleUUID();

	protected ConnectionFactory mFactory;
	protected Connection mConnection;
	protected Channel mChannel;

	protected String exchange;
	protected String routingKey;
	protected Integer prefetchSize;
	protected Integer prefetchCount;

	@Override
	public String getExchange() {
		return exchange;
	}

	@Override
	public String getRoutingKey() {
		return routingKey;
	}

	@Override
	public void initialize(AppConfig config) throws Exception {
		this.mFactory = new ConnectionFactory();
		this.mFactory.setHost(config.getString("rabbit.host"));
		this.mFactory.setPort(config.getInteger("rabbit.port"));
		this.mFactory.setUsername(config.getString("rabbit.username"));
		this.mFactory.setPassword(config.getString("rabbit.password"));

		// 获取当前支持的指令
		this.exchange = config.getString("rabbit.exchange");
		this.routingKey = config.getString("rabbit.routings");
		this.prefetchSize = config.getInteger("rabbit.prefetch.size");
		this.prefetchCount = config.getInteger("rabbit.prefetch.count");

		// 创建一个连接
		this.mConnection = mFactory.newConnection();

		// 创建一个通道
		this.mChannel = mConnection.createChannel();
		this.mChannel.exchangeDeclare(exchange, "direct", false, false, null);
		this.mChannel.queueDeclare(queue, false, false, true, null);
		this.mChannel.queueBind(queue, exchange, routingKey);

		// 队列绑定到路由上
		this.mChannel.basicQos(prefetchSize, prefetchCount, true);
		this.mChannel.basicConsume(queue, false, new RabbitConsumer(this));
	}

	@Override
	public boolean send(Packet packet) {
		try {
			for (RabbitConsumerHandler<?> handler : mHandlers) {
				if (handler.isSupport(packet.cmd())) {//
					return handler.doHandler0(packet);
				}
			}

			return false;
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}

		return false;
	}

	@Override
	public boolean sendRabbit(Packet packet) {
		return this.sendRabbit(routingKey, packet);
	}

	@Override
	public boolean sendRabbit(String routingKey, Packet packet) {
		return this.sendRabbit(exchange, routingKey, packet);
	}

	@Override
	public boolean sendRabbit(String exchange, String routingKey, Packet packet) {
		try {
			// 构建应答头部信息
			Map<String, Object> replyHeaders = new HashMap<>();
			replyHeaders.put(RabbitHeaders.RABBIT_ROUTING_KEY, routingKey);
			replyHeaders.put(RabbitHeaders.RABBIT_EXCHANGE, exchange);
			replyHeaders.put(RabbitHeaders.MAC, packet.getAddress());
			replyHeaders.put(RabbitHeaders.CMD, packet.cmd());

			// 构建应答消息属性
			BasicProperties.Builder builder = new BasicProperties.Builder();
			builder.contentType("application/json");
			builder.contentEncoding("utf-8");
			builder.headers(replyHeaders);

			PacketReply packetReply = new PacketReply();

			BasicProperties properties1 = builder.build();
			byte[] packetBytes = packetReply.serialize(charset);

			// 发送回复内容
			this.mChannel.basicPublish(exchange, routingKey, properties1, packetBytes);

			return true;
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}

		return false;
	}

	@Override
	public boolean sendReply(RabbitConsumerProps props, PacketReply packetReply) {
		packetReply.setRabbitRoutingKey(routingKey);
		packetReply.setRabbitExchange(exchange);

		try {
			// 构建应答头部信息
			Map<String, Object> replyHeaders = new HashMap<>();
			replyHeaders.put(RabbitHeaders.CMD, props.getCmd());
			replyHeaders.put(RabbitHeaders.MAC, props.getMac());

			// 构建应答消息属性
			BasicProperties.Builder builder = new BasicProperties.Builder();
			builder.correlationId(props.getCorrelationId());
			builder.contentType("application/json");
			builder.contentEncoding("utf-8");
			builder.headers(replyHeaders);

			String routingKey = props.getReplyTo();
			BasicProperties properties1 = builder.build();
			byte[] massageBytes = packetReply.serialize(charset);

			// 发送回复内容
			mChannel.basicPublish("", routingKey, properties1, massageBytes);

			return true;
		} catch (Exception e) {
			logger.error("应答消息异常：" + e.getMessage());
		}

		return false;
	}

	public void addHandler(RabbitConsumerHandler<?> handler) {
		this.mHandlers.add(handler);
	}

	@Override
	public void destroy() throws Exception {
		this.mConnection.close();
		this.mChannel.close();
	}

}
